package com.lry.service.impl;

import com.lry.core.Log;
import com.lry.core.RaftNode;
import com.lry.entity.AppendEntriesRequest;
import com.lry.entity.AppendEntriesResponse;
import com.lry.entity.VoteRequest;
import com.lry.entity.VoteResponse;
import com.lry.service.RaftConsensusService;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class RaftConsensusServiceImpl implements RaftConsensusService {

    private RaftNode raftNode;

    public RaftConsensusServiceImpl(RaftNode raftNode){
        this.raftNode = raftNode;
    }

    @Override
    public VoteResponse requestVote(VoteRequest voteRequest) {
        //如果我的term大于请求的term，则拒绝
        if(raftNode.getCurrentTerm()>voteRequest.getTerm()){
            return new VoteResponse(raftNode.getCurrentTerm(),false);
        }
        if(null!=raftNode.getVotedFor()){
            return new VoteResponse(raftNode.getCurrentTerm(),false);
        }
        raftNode.setVotedFor(voteRequest.getCandidateId());
        return new VoteResponse(raftNode.getCurrentTerm(),true);
    }

    @Override
    public AppendEntriesResponse appendEntries(AppendEntriesRequest appendEntriesRequest) {
        log.info("server={},收到leaderId={}的 appendEntries",raftNode.getServerId(),appendEntriesRequest.getLeaderId());

        AppendEntriesResponse response = new AppendEntriesResponse(raftNode.getCurrentTerm(),false);

        if (null==raftNode.getLeaderId()) {
            raftNode.setLeaderId(appendEntriesRequest.getLeaderId());
        }

        //收到leader的追加日志或者心跳，重置自己的选举计时器
        raftNode.resetElectionTimer(); //先注释掉，不给他篡权，为了调试日志方便

        //如果我的term大于请求的term，则拒绝
        if(raftNode.getCurrentTerm()>appendEntriesRequest.getTerm()){
            return response;
        }

        if (raftNode.getLeaderId() != appendEntriesRequest.getLeaderId()) {
            log.warn("Another peer={} declares that it is the leader " +
                            "at term={} which was occupied by leader={}",
                    appendEntriesRequest.getLeaderId(), appendEntriesRequest.getTerm(), raftNode.getLeaderId());
            raftNode.stepDown(appendEntriesRequest.getTerm() + 1);
            return new AppendEntriesResponse(appendEntriesRequest.getTerm() + 1,false);
        }

        if (appendEntriesRequest.getPreLogIndex() > raftNode.getRaftLog().getLastLogIndex()) {
            log.info("Rejecting AppendEntries RPC would leave gap, " +
                            "request prevLogIndex={}, my lastLogIndex={}",
                    appendEntriesRequest.getPreLogIndex(), raftNode.getRaftLog().getLastLogIndex());
            return response;
        }

        //心跳
        if(appendEntriesRequest.getEntries()==null || appendEntriesRequest.getEntries().isEmpty()){
            return new AppendEntriesResponse(raftNode.getCurrentTerm(),true);
        }

        long preLogIndex = appendEntriesRequest.getPreLogIndex();
        long preLogTerm = appendEntriesRequest.getPreLogTerm();
        List<Log> logs = appendEntriesRequest.getEntries();


        //在接收者的日志中查找看看是否可以找到和pre一样的日志，没有的话，说明当前节点数据落后了，需要返回false，等待leader追加过来
        boolean findPre = false;
        Log entry = raftNode.getRaftLog().getEntry(preLogIndex);
        if(null!=entry && entry.getTerm()==preLogTerm){
            findPre = true;
        }
        if(!findPre){
            return new AppendEntriesResponse(raftNode.getCurrentTerm(),false);
        }


        //如果一个已经存在的条目和新条目发生了冲突（索引相同，任期不同），那么就删除这个已经存在条目及它之后的所有条目
        for (Log log1 : logs) {
            entry = raftNode.getRaftLog().getEntry(log1.getIndex());
            if(null==entry){
                continue;
            }
            if(entry.getTerm()!=log1.getTerm()){
                raftNode.getRaftLog().truncateSuffix(log1.getIndex());
                break;
            }
        }

        //追加日志条目
        for (Log log1 : logs) {
            log.info("server={},追加日志={}",raftNode.getServerId(),new String(log1.getCommand()));
            raftNode.getRaftLog().append(log1);
        }

        advanceCommitIndex(appendEntriesRequest);

        return new AppendEntriesResponse(raftNode.getCurrentTerm(),true);
    }

    public void advanceCommitIndex(AppendEntriesRequest request){

        long newCommitIdx = Math.min(request.getLeaderCommit(),request.getPreLogIndex()+request.getEntries().size());
        raftNode.setCommitIndex(newCommitIdx);

        if(raftNode.getLastApplied()<raftNode.getCommitIndex()){
            for(long idx = raftNode.getLastApplied();idx<=raftNode.getCommitIndex();idx++){
                Log entry = raftNode.getRaftLog().getEntry(idx);
                if(null==entry){
                    continue;
                }
                raftNode.getStateMachine().apply(entry.getCommand());
                raftNode.setLastApplied(idx);
            }
        }
    }

}
